﻿using System;
using System.Collections.Generic;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQDemo.Common;

namespace RabbitMQDemo.Consumer
{
    public class DeadLetter
    {
        private const string ExchangeName = "dead_letter_exchange";

        private const string QueueName = "dead_letter_queue";

        private const string RoutingKey = "dead_letter_routing";

        private const string DLX_ExchangeName = "dead_letter_exchange_dlx";

        private const string DLX_QueueName = "dead_letter_queue_dlx";

        private const string DLX_RoutingKey = "dead_letter_routing_dlx";

        public static void Receive(string[] args)
        {
            var isReceive = Utils.GetFirstValue(args) == 0;

            using var connection = MQClient.GetConnection();
            using var channel = connection.CreateModel();

            // 声明死信交换机、死信队列
            channel.ExchangeDeclare(exchange: DLX_ExchangeName, type: "direct", durable: false, autoDelete: false, arguments: null);
            channel.QueueDeclare(queue: DLX_QueueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
            channel.QueueBind(queue: DLX_QueueName, exchange: DLX_ExchangeName, routingKey: DLX_RoutingKey, arguments: null);

            // 声明交换机、队列
            channel.ExchangeDeclare(exchange: ExchangeName, type: "direct", durable: false, autoDelete: false, arguments: null);
            channel.QueueDeclare(queue: QueueName, durable: false, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> {
                { "x-dead-letter-exchange", DLX_ExchangeName }, // 设置当前队列的死信交换机DLX
                { "x-dead-letter-routing-key", DLX_RoutingKey }, // 设置死信路由key，DLX会根据该值去找到死信消息存放的队列
                //{ "x-expires", 10000 }, // 队列过期时间
                { "x-message-ttl", 6000 } // 消息过期时间
            });
            channel.QueueBind(queue: QueueName, exchange: ExchangeName, routingKey: RoutingKey, arguments: null);

            // 死信队列消费
            DLXReceive(channel);

            // 业务消息消费
            using var newChannel = connection.CreateModel();
            if (isReceive)
            {
                Receive(newChannel);
            }

            Console.WriteLine("按任意键退出");
            Console.ReadKey();
        }

        public static void Receive(IModel channel)
        {
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());

                if (ea.DeliveryTag % 2 == 0)
                {
                    channel.BasicNack(ea.DeliveryTag, false, false);
                    Console.WriteLine($"[DeadLetter]received：{message}. Nack and requeue=false");
                }
                else
                {
                    channel.BasicReject(ea.DeliveryTag, false);
                    Console.WriteLine($"[DeadLetter]received：{message}. Reject and requeue=false");
                }
            };

            channel.BasicConsume(queue: QueueName, autoAck: false, consumerTag: "testConsumer", noLocal: false, exclusive: false, arguments: null, consumer: consumer);
        }

        public static void DLXReceive(IModel channel)
        {
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var headerKeys = new string[] { "x-first-death-exchange", "x-first-death-queue", "x-first-death-reason" };
                var headers = new StringBuilder();
                foreach (var item in headerKeys)
                {
                    if (ea.BasicProperties.Headers.ContainsKey(item))
                        headers.AppendLine($"{item}={Encoding.UTF8.GetString((byte[])ea.BasicProperties.Headers[item])}");
                }

                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                Console.WriteLine($"[DLX]received：{message}. Exchange={ea.Exchange}, routingKey={ea.RoutingKey}, header={headers}");
            };

            channel.BasicConsume(queue: DLX_QueueName, autoAck: true, consumerTag: "testConsumer", noLocal: false, exclusive: false, arguments: null, consumer: consumer);
        }
    }
}
